package com.navinfo.platform.dataanalysis.service.impl;

import com.navinfo.platform.common.dto.DataInfo;
import com.navinfo.platform.dataanalysis.dto.IdleBean;
import com.navinfo.platform.dataanalysis.service.IIdleSpeedService;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@SuppressWarnings("Duplicates")
public enum IdleSpeedService implements IIdleSpeedService {
    /***
     *
     */
    INSTANCE;

    @Override
    public JavaRDD<IdleBean> calcIdleInfo(JavaPairRDD<Long, List<DataInfo>> dataInfoRdd) {
        return dataInfoRdd.mapPartitionsToPair(d -> {
            List<Tuple2<Long, IdleBean>> idleBeans = new ArrayList<>();
            Map<Long, IdleBean> tempDataMap = new HashMap<>();
            while(d.hasNext()){
                Tuple2<Long, List<DataInfo>> tuple2 = d.next();
                long tid = tuple2._1();
                List<DataInfo> dataInfos = tuple2._2();
                IdleBean idleBean = null;
                int size = dataInfos.size();
                boolean lastPoint = false;
                int count = 1;
                for(DataInfo dataInfo : dataInfos){
                    if(count == size){
                        lastPoint = true;
                    }
                    idleBean = calcIdle(tempDataMap, tid, lastPoint, dataInfo);
                    count++;
                }
                if (idleBean == null) { continue; }
                idleBean.setIdle100kmTimes(idleBean.getIdleTimes());
                idleBeans.add(Tuple2.apply(idleBean.getTid(), idleBean));
            }
            return idleBeans.iterator();
            }).mapPartitions(d -> {
                List<IdleBean> idleBeans = new ArrayList<>();
                while(d.hasNext()){
                    Tuple2<Long, IdleBean> tuple2 = d.next();
                    idleBeans.add(tuple2._2());
                }
                return idleBeans.iterator();
            });
    }

    private IdleBean calcIdle(Map<Long, IdleBean> tempDataMap, long tid, boolean lastPoint, DataInfo dataInfo) {
        IdleBean idleBean = tempDataMap.get(tid);
        if (idleBean == null) {
            idleBean = new IdleBean();
            idleBean.setTid(tid);
            tempDataMap.put(tid, idleBean);
        }
        int rotation = dataInfo.getRotation();
        int speedKmh = dataInfo.getSpeed();
        int gpsTime = (int)dataInfo.getGpsDate();
        boolean condition = !lastPoint && rotation > 0 && speedKmh == 0 && idleBean.calcDuration(gpsTime) < 120;
        if (condition) {
            if (idleBean.getFirstGpsTime() == 0) {
                idleBean.setFirstGpsTime(gpsTime);
            }
        } else {
            int firstGpstime = idleBean.getFirstGpsTime();
            if (firstGpstime != 0) {
                int duration = gpsTime - firstGpstime;
                idleBean.incrIdleDuration(duration);
                idleBean.incrIdleTimes();
                idleBean.clearData();
            }
        }
        return idleBean;
    }
}
